package com.boot.test;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

import java.sql.Time;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class RxJavaOperators {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch finish = new CountDownLatch(1);
//        //create
//        Observable.create(observableEmitter ->{
//            if (!observableEmitter.isDisposed()) {
//                for (int i=0;i<3;i++) {
//                    observableEmitter.onNext(i);
//                }
//                observableEmitter.onComplete();
//            }
//        }).subscribe(System.out::println, error -> error.printStackTrace(), () -> System.out.println("Complete"));
//        //just
//        Observable.just(1, 2, 3, 4, 5, 6)
//                .subscribe(System.out::println, error -> error.printStackTrace(), () -> System.out.println("Complete"));
//        //from
//        Observable.fromArray("hello","world")
//                .subscribe(System.out::println, error -> error.printStackTrace(), () -> System.out.println("Complete"));
//        List<String> list = Arrays.asList("hello","world");
//        Observable.fromIterable(list).subscribe(System.out::println, error -> error.printStackTrace(), () -> System.out.println("Complete"));
//
//        ExecutorService executorService = Executors.newCachedThreadPool();
//        Future<String> future= executorService.submit(()->{
//            TimeUnit.SECONDS.sleep(1);
//            return "Ok";
//        });
//        Observable.fromFuture(future)
//                .subscribe(System.out::println, error -> error.printStackTrace(), () -> System.out.println("Complete"));
//        //repeat

//        Observable.just(100).repeat(3)
//                .subscribe(System.out::println);
//        //repeatWhen
//        Observable.just(20).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
//            int count = 0;
//            @Override
//            public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
//                return objectObservable.flatMap(o->{
//                    if(++count > 3) {
//                        //return Observable.empty();
//                        return Observable.error(new Throwable("finish repeat"));
//                    }
//                    return Observable.timer(2+count, TimeUnit.SECONDS);
//                });
//            }
//        }).subscribe(System.out::println, error->finish.countDown(),finish::countDown);
//        //repeatUntil
//        long startTime = System.currentTimeMillis();
//        Observable.interval(1,TimeUnit.SECONDS).take(3).repeatUntil(() -> System.currentTimeMillis() - startTime > 5000)
//                .subscribe(System.out::println, error -> finish.countDown(),() -> {System.out.println("Completed");});
//        //Defer
//        Consumer subscribe1 = l -> System.out.println("subscriber1: " + l);
//        Consumer subscribe2 = l -> System.out.println("subscriber1: " + l);
//        Observable observable = Observable.defer(() -> Observable.range(0, 3));
//        observable.subscribe(subscribe1, error -> finish.countDown(), finish::countDown);
//        observable.delaySubscription(1, TimeUnit.SECONDS)
//        .subscribe(subscribe2, error -> finish.countDown(), finish::countDown);
//
//        //interval
        Observable.just(1,2,3).interval(1,TimeUnit.SECONDS).take(3).subscribe(System.out::println,error -> finish.countDown(),() -> System.out.println("completed"));
        //timer
//        Observable.timer(1,TimeUnit.SECONDS).subscribe(System.out::println,error -> finish.countDown(),() -> System.out.println("completed"));
    }

}
